package top.tenebrous.termbor.rpc.consumer.manager;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.tenebrous.termbor.rpc.common.helper.RpcServiceHelper;
import top.tenebrous.termbor.rpc.common.pool.ClientThreadPool;
import top.tenebrous.termbor.rpc.consumer.handler.RpcConsumerHandler;
import top.tenebrous.termbor.rpc.consumer.helper.RpcConsumerHandlerHelper;
import top.tenebrous.termbor.rpc.consumer.initializer.RpcConsumerInitializer;
import top.tenebrous.termbor.rpc.protocol.RpcProtocol;
import top.tenebrous.termbor.rpc.protocol.message.RpcRequest;
import top.tenebrous.termbor.rpc.protocol.meta.ServiceMeta;
import top.tenebrous.termbor.rpc.proxy.consumer.Consumer;
import top.tenebrous.termbor.rpc.proxy.future.RPCFuture;
import top.tenebrous.termbor.rpc.registry.RegistryService;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: termbor-rpc
 * @description:
 * @author: yazhang6
 * @create: 2024-05-22 11:15
 **/
public class RpcConsumerManager implements Consumer {

    private final Logger logger = LoggerFactory.getLogger(RpcConsumerManager.class);

    private final Bootstrap bootstrap;

    private final EventLoopGroup eventLoopGroup;

    private static volatile RpcConsumerManager instance;

    private static Map<String, RpcConsumerHandler> handlerMap = new ConcurrentHashMap<>();

    private RpcConsumerManager() {
        this.bootstrap = new Bootstrap();
        this.eventLoopGroup = new NioEventLoopGroup(4);
        this.bootstrap.group(this.eventLoopGroup).channel(NioSocketChannel.class)
            .handler(new RpcConsumerInitializer());
    }

    public static RpcConsumerManager getInstance() {
        if (instance == null) {
            synchronized (RpcConsumerManager.class) {
                if (instance == null) {
                    instance = new RpcConsumerManager();
                }
            }
        }
        return instance;
    }

    public void close() {
        RpcConsumerHandlerHelper.closeRpcClientHandler();
        eventLoopGroup.shutdownGracefully();
        ClientThreadPool.shutdown();
    }

    @SneakyThrows
    @Override
    public RPCFuture sendRequest(RpcProtocol<RpcRequest> protocol, RegistryService registryService) {
        RpcRequest request = protocol.getBody();
        String serviceKey = RpcServiceHelper.buildServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        Object[] params = request.getParameters();

        int invokerHashCode = (params == null || params.length <= 0) ? serviceKey.hashCode() : params[0].hashCode();
        ServiceMeta serviceMeta = registryService.discovery(serviceKey, invokerHashCode);
        if (serviceMeta != null) {
            RpcConsumerHandler handler = RpcConsumerHandlerHelper.get(serviceMeta);
            if (handler == null) {
                handler = getRpcConsumerHandler(serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
                RpcConsumerHandlerHelper.put(serviceMeta, handler);
            } else if (!handler.getChannel().isActive()) {      // 缓存中存在RpcClientHandler，但不活跃
                handler.close();
                handler = getRpcConsumerHandler(serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
                RpcConsumerHandlerHelper.put(serviceMeta, handler);
            }
            return handler.sendRequest(protocol, request.isAsync(), request.isOneway());
        }
        return null;

    }

    @SneakyThrows
    private RpcConsumerHandler getRpcConsumerHandler(String serviceAddress, int port) {
        ChannelFuture channelFuture = this.bootstrap.connect(serviceAddress, port).sync();
        channelFuture.addListener((ChannelFutureListener) listener -> {
            if (listener.isSuccess()) {
                logger.info("connect rpc server {} on port {} success.", serviceAddress, port);
            } else {
                logger.error("connect rpc server {} on port {} failed.", serviceAddress, port);

                channelFuture.cause().printStackTrace();
                eventLoopGroup.shutdownGracefully();
            }
        });
        return channelFuture.channel().pipeline().get(RpcConsumerHandler.class);
    }

}
